package com.om.opensourway;

import Utils.HttpClientUtils;
import org.apache.http.HttpHeaders;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.client.utils.URIBuilder;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.util.EntityUtils;
import org.apache.log4j.Logger;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.map.ObjectMapper;

import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Base64;
import java.util.Map;
import java.util.Properties;
/**
 * @author zhxia
 * @date 2020/9/28 16:13
 */

public class NginxErrorDataBolt extends BaseRichBolt {
    static Properties conf = new Properties();
    public static CloseableHttpClient httpClient;
    static Logger logger;
    static String NGINXINDEX;
    static ObjectMapper objectMapper=new ObjectMapper();
    static {
        try {
            logger = org.apache.log4j.Logger.getLogger("STDERR");
            httpClient = HttpClientUtils.getClient();
            InputStream resourceAsStream = AnalysisForStreaming.class.getResourceAsStream("/resources/conf.properties");
            conf.load(resourceAsStream);
            NGINXINDEX=conf.getProperty("nginx.error.index");
        }
         catch (IOException e) {
            e.printStackTrace();
        }
    }

    URI uri = new URIBuilder().setScheme(conf.getProperty("es.scheme")).setHost(conf.getProperty("es.host")).setPort(Integer.parseInt(conf.getProperty("es.port"))).setPath("/_bulk").build();

    private OutputCollector collector;

    public NginxErrorDataBolt() throws URISyntaxException {
    }

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;

    }

    @Override
    public void execute(Tuple input) {
        try {
            String value = input.getStringByField("value");
            JsonNode originalData=objectMapper.readTree(value);
            JsonNode sourceDataNode =originalData .get("_source");
            String logstr = sourceDataNode.get("log").getTextValue();
            boolean isRight=isOk((logstr));
            if(!isRight) {
                String result = assemblydata(sourceDataNode);
                String indexstr = "{\"index\":{\"_index\":\"" + NGINXINDEX + "\",\"_id\":\"" + originalData.get("_id").getTextValue() + "\"}";
                StringBuffer resultBuffer = new StringBuffer();
                resultBuffer.append(indexstr).append("\n").append(result).append("\n");
                CloseableHttpResponse execute = null;
                try {
                    HttpPost httpPost = new HttpPost(uri);
                    httpPost.setEntity(new StringEntity(resultBuffer.toString(), "UTF-8"));
                    httpPost.addHeader(HttpHeaders.CONTENT_TYPE, "application/x-ndjson");
                    httpPost.addHeader("Authorization", "Basic " + Base64.getEncoder().encodeToString((conf.getProperty("es.user") + ":" + conf.get("es.passwd")).getBytes()));
                    execute = httpClient.execute(httpPost);
                } catch (Exception e) {
                    logger.error(e.getMessage());
                    logger.error(EntityUtils.toString(execute.getEntity()), e);
                } finally {
                    if (execute != null) {
                        EntityUtils.consume(execute.getEntity());
                    }
                }
            }
            collector.ack(input);
        } catch (Exception e) {
            collector.fail(input);
        }
    }
    public static String assemblydata(JsonNode sourceDataNode) throws IOException {
        String result = objectMapper.writeValueAsString(sourceDataNode.get("_source"));
        return result;
    }
    public boolean isOk(String string){
        if(string.startsWith("{\\\"time\\")){
            return  true;
        }
        else {
            return  false;
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}
